package com.ichoice.rabbitmq.rpc.helper.client;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.ichoice.rabbitmq.rpc.helper.consts.CommonContant;
import com.ichoice.rabbitmq.rpc.helper.model.SubScribeModel;
import com.ichoice.rabbitmq.rpc.helper.sub.SubScribeStoreHelper;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * 绑定服务
 *
 * @author xianrensawang
 * @since 2021/1/28
 */
@Slf4j
public class ClientBlinding {

    private final CachingConnectionFactory cachingConnectionFactory;

    public ClientBlinding(CachingConnectionFactory cachingConnectionFactory) {
        this.cachingConnectionFactory = cachingConnectionFactory;
    }

    /**
     * 绑定当前客户端队列(区分分布式，ip+port)
     *
     * @param exchange 交换机名称
     * @param key      路由键
     * @param queue    队列
     */
    public void blind(String exchange, String key, String queue) {
        cachingConnectionFactory.createConnection();
        Connection connection = cachingConnectionFactory.createConnection();
        final Channel channel = connection.createChannel(false);
        try {
            channel.exchangeDeclare(exchange, "topic", true);
            channel.queueDeclare(queue, true, false, false, null);
            channel.queueBind(queue, exchange, key);
            // 只绑定关系,消费者在ExecuteBlinding里实现
        } catch (IOException e) {
            log.error("channel添加队列失败" + queue);
        }
    }

    /**
     * 绑定广播交换机,绑定当前实例队列,监听客户端发布订阅信息并缓存
     *
     * @param exchange 广播交换机
     * @param queue    广播队列
     * @param key fanout Key
     */
    public void blindFanout(String exchange, String queue,String key) {
        cachingConnectionFactory.createConnection();
        Connection connection = cachingConnectionFactory.createConnection();
        final Channel channel = connection.createChannel(false);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    JSONObject jsonObject = JSONUtil.parseObj(new String(body, StandardCharsets.UTF_8));
                    SubScribeModel scribeModel = JSONUtil.toBean(jsonObject, SubScribeModel.class);
                    if(Objects.equals(scribeModel.getDel(),"1")){
                        SubScribeStoreHelper.remove(scribeModel);
                        log.info("服务端移除来自客户端的发布订阅信息成功,客户端:{},服务端bean:{},服务端方法:{}",
                                scribeModel.getClientName(), scribeModel.getServerBean(), scribeModel.getServerMethod());
                    }
                    if(Objects.equals(scribeModel.getDel(),"0")){
                        SubScribeStoreHelper.store(scribeModel);
                        log.info("服务端缓存来自客户端的发布订阅信息成功,客户端:{},服务端bean:{},服务端方法:{}",
                                scribeModel.getClientName(), scribeModel.getServerBean(), scribeModel.getServerMethod());
                    }
                } catch (Exception e) {
                    log.error(e.getMessage());
                }
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        try {
            // 配置延迟消息队列
            Map<String, Object> exchangeConfig = new HashMap<>();
            exchangeConfig.put("x-delayed-type", "direct");
            String fanoutQueue = queue + "-fanout";
            channel.exchangeDeclare(CommonContant.EXCHANGE_DELAY, "x-delayed-message", true, false, exchangeConfig);
            channel.queueDeclare(fanoutQueue, true, false, false, null);
            channel.queueBind(fanoutQueue, CommonContant.EXCHANGE_DELAY, key);
            // 改成手动确认
            channel.basicConsume(fanoutQueue, false, consumer);
        } catch (IOException e) {
            log.error("channel添加广播队列失败" + queue);
        }
    }
}
